package com.shujia.rec.funaction

import com.shujia.rec.common.Constants
import com.shujia.rec.entry.CaseClass.CountEntry
import org.apache.flink.api.common.state.ListStateDescriptor
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector
import org.apache.flink.api.common.state.ListState

import scala.collection.mutable.ListBuffer

/**
  * 统计每个窗口中前10个商品
  *
  */
class TopNKeyedProcessFunction extends KeyedProcessFunction[Long, CountEntry, List[CountEntry]] {

  //自定义状态存储top  商品
  var topNState: ListState[CountEntry] = _

  override def open(parameters: Configuration): Unit = {
    val desc = new ListStateDescriptor[CountEntry]("topNState", classOf[CountEntry])
    topNState = getRuntimeContext.getListState(desc)
  }

  override def processElement(value: CountEntry, ctx: KeyedProcessFunction[Long, CountEntry, List[CountEntry]]#Context, out: Collector[List[CountEntry]]): Unit = {

    topNState.add(value)

    //注册定时器定时器
    ctx.timerService().registerEventTimeTimer(value.end + 5000)


  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, CountEntry, List[CountEntry]]#OnTimerContext, out: Collector[List[CountEntry]]): Unit = {


    val list = new ListBuffer[CountEntry]

    //取top10 商品
    val iter = topNState.get().iterator()
    while (iter.hasNext) {
      list.+=(iter.next())
    }

    //清空状态数据
    topNState.clear()

    //点击量倒叙排序
    val topN = list.sortBy(-_.count).take(Constants.LOGTOHBASE_TOP_SIZE)

    //返回
    out.collect(topN.toList)
  }
}
